# @Version        : 1.0
# @Update Time    : 2025/8/21 0:21
# @File           : send_mail.py
# @IDE            : PyCharm
# @Desc           : 更优雅的邮件发送封装（支持文本/HTML/模板/附件/内联图片）

import asyncio
from pathlib import Path
from typing import Any, Dict, List, Optional, Sequence, Union

import fastapi_mail.errors
from fastapi_mail import ConnectionConfig, FastMail, MessageSchema, MessageType
from pydantic import BaseModel, DirectoryPath, EmailStr, Field, ValidationError

from applications import settings
from applications.log import logger


# -------------------------
# 配置与结果模型
# -------------------------
class EmailConfig(BaseModel):
    # FastAPI-Mail 典型字段，允许额外字段透传
    MAIL_USERNAME: Optional[str] = None
    MAIL_PASSWORD: Optional[str] = None
    MAIL_FROM: Optional[EmailStr] = None
    MAIL_PORT: Optional[int] = None
    MAIL_SERVER: Optional[str] = None
    MAIL_FROM_NAME: Optional[str] = None
    USE_CREDENTIALS: bool = True
    VALIDATE_CERTS: bool = True
    MAIL_STARTTLS: Optional[bool] = None
    MAIL_SSL_TLS: Optional[bool] = None
    SUPPRESS_SEND: int = 0  # 1 表示仅模拟发送（不真正发出）
    TEMPLATE_FOLDER: Optional[DirectoryPath] = None

    class Config:
        extra = "allow"  # 允许扩展字段

    def to_connection_config(self) -> ConnectionConfig:
        return ConnectionConfig(**self.model_dump())


class EmailResult(BaseModel):
    ok: bool = False
    message: str = ""
    subject: str = ""
    recipients: List[str] = Field(default_factory=list)
    error: Optional[str] = None

    def to_legacy(self) -> Dict[str, Any]:
        # 兼容老的 {"code": 1|0, "message": "..."} 返回
        return {"code": 1 if self.ok else 0, "message": self.message}


# -------------------------
# 工具函数
# -------------------------
def _to_subtype(subtype: Union[str, MessageType, None]) -> MessageType:
    if isinstance(subtype, MessageType):
        return subtype
    if isinstance(subtype, str) and subtype.lower() == "plain":
        return MessageType.plain
    return MessageType.html


def _normalize_recipients(recipients: Union[str, EmailStr, Sequence[Union[str, EmailStr]]]) -> List[str]:
    if isinstance(recipients, (str, EmailStr)):
        return [str(recipients)]
    return [str(r) for r in recipients if r]


def build_inline_image(
    path: Union[str, Path],
    cid: str = "logo",
    filename: Optional[str] = None,
    mime_subtype: str = "png",
) -> Dict[str, Any]:
    """
    构建内联图片附件，模板中可通过 <img src="cid:logo"> 使用。
    """
    p = Path(path)
    filename = filename or p.name
    headers = {
        "Content-ID": f"<{cid}>",
        "Content-Disposition": f'inline; filename="{filename}"',
    }
    return {
        "file": str(p),
        "headers": headers,
        "mime_type": "image",
        "mime_subtype": mime_subtype,
    }


# -------------------------
# 发送服务
# -------------------------
class EmailService:
    def __init__(
        self,
        email_cfg: Union[EmailConfig, Dict[str, Any], ConnectionConfig, None] = None,
        *,
        template_folder: Optional[DirectoryPath] = None,
        default_context: Optional[Dict[str, Any]] = None,
        subject_prefix: Optional[str] = None,
        inline_logo_path: Optional[Union[str, Path]] = None,
    ):
        """
        email_cfg: 可传 dict / EmailConfig / ConnectionConfig。若不传，会尝试使用 settings.email.dict()
        template_folder: 模板目录（优先级高于 email_cfg.TEMPLATE_FOLDER）
        default_context: 默认模板上下文（每次发送会 merge）
        subject_prefix: 主题前缀，如 "[系统通知] "（会自动拼在 subject 前）
        inline_logo_path: 设置后在模板邮件中自动附加内联图片 -> cid: logo
        """
        self._conn_cfg = self._build_connection_config(email_cfg, template_folder)
        self._fm = FastMail(self._conn_cfg)
        self._default_context = default_context or {}
        self._subject_prefix = subject_prefix or ""
        self._inline_logo_path = Path(inline_logo_path) if inline_logo_path else None

    def _build_connection_config(
        self,
        email_cfg: Union[EmailConfig, Dict[str, Any], ConnectionConfig, None],
        template_folder: Optional[DirectoryPath],
    ) -> ConnectionConfig:
        if isinstance(email_cfg, ConnectionConfig):
            cfg = email_cfg
        else:
            if email_cfg is None:
                # 回落到项目内全局配置
                raw = settings.MAIL_CONFIG
            elif isinstance(email_cfg, EmailConfig):
                raw = email_cfg.model_dump()
            else:
                raw = dict(email_cfg)

            # 覆盖模板目录
            if template_folder:
                raw["TEMPLATE_FOLDER"] = template_folder

            try:
                cfg = EmailConfig(**raw).to_connection_config()
            except ValidationError as e:
                raise ValueError(f"邮件配置无效: {e}") from e

        # 基本校验
        if not getattr(cfg, "MAIL_SERVER", None):
            raise ValueError("请配置有效的 MAIL_SERVER")
        return cfg

    def _build_subject(self, subject: str) -> str:
        return f"{self._subject_prefix}{subject}" if self._subject_prefix else subject

    def _merge_context(self, ctx: Optional[Dict[str, Any]]) -> Dict[str, Any]:
        merged = {**self._default_context}
        if ctx:
            merged.update(ctx)
        return merged

    def _auto_attachments_for_template(self) -> List[Dict[str, Any]]:
        attachments: List[Dict[str, Any]] = []
        if self._inline_logo_path and self._inline_logo_path.exists():
            attachments.append(build_inline_image(self._inline_logo_path, cid="logo"))
        return attachments

    def _build_message(
        self,
        *,
        subject: str,
        recipients: Union[str, EmailStr, Sequence[Union[str, EmailStr]]],
        subtype: Union[str, MessageType, None] = "html",
        body: Optional[str] = None,
        template_body: Optional[Dict[str, Any]] = None,
        attachments: Optional[List[Dict[str, Any]]] = None,
        headers: Optional[Dict[str, str]] = None,
        **kwargs: Any,
    ) -> MessageSchema:
        if not body and not template_body:
            raise ValueError("必须提供 body 或 template_body 其中之一")

        msg_kwargs: Dict[str, Any] = {
            "subject": self._build_subject(subject),
            "recipients": _normalize_recipients(recipients),
            "subtype": _to_subtype(subtype),
        }

        if body:
            msg_kwargs["body"] = body
        if template_body:
            msg_kwargs["template_body"] = template_body

        if attachments:
            msg_kwargs["attachments"] = attachments
        if headers:
            msg_kwargs["headers"] = headers

        # 透传其他 fastapi-mail 支持的字段（如 cc / bcc / reply_to / alternatives 等）
        msg_kwargs.update(kwargs)

        return MessageSchema(**msg_kwargs)

    async def send(
        self,
        *,
        subject: str,
        recipients: Union[str, EmailStr, Sequence[Union[str, EmailStr]]],
        subtype: Union[str, MessageType, None] = "html",
        body: Optional[str] = None,
        template_name: Optional[str] = None,
        template_context: Optional[Dict[str, Any]] = None,
        attachments: Optional[List[Dict[str, Any]]] = None,
        headers: Optional[Dict[str, str]] = None,
        timeout: Optional[float] = None,
        add_default_inline_logo: bool = True,
        **kwargs: Any,
    ) -> EmailResult:
        """
        通用发送方法：既可发纯文本/HTML，也可发模板邮件。
        """
        try:
            if template_name:
                # 模板邮件
                context = self._merge_context(template_context)
                if add_default_inline_logo:
                    auto_atts = self._auto_attachments_for_template()
                    if auto_atts:
                        attachments = (attachments or []) + auto_atts

                message = self._build_message(
                    subject=subject,
                    recipients=recipients,
                    subtype=subtype,
                    template_body=context,
                    attachments=attachments,
                    headers=headers,
                    **kwargs,
                )
                coro = self._fm.send_message(message, template_name=template_name)
            else:
                # 纯文本 / HTML
                message = self._build_message(
                    subject=subject,
                    recipients=recipients,
                    subtype=subtype,
                    body=body or "",
                    attachments=attachments,
                    headers=headers,
                    **kwargs,
                )
                coro = self._fm.send_message(message)

            if timeout:
                await asyncio.wait_for(coro, timeout=timeout)
            else:
                await coro

            res = EmailResult(
                ok=True,
                message="发送成功",
                subject=self._build_subject(subject),
                recipients=_normalize_recipients(recipients),
            )
            logger.success(f"邮件发送成功 | 接收人 - {res.recipients} | 主题 - {res.subject}")
            return res

        except asyncio.TimeoutError as e:
            msg = f"发送超时: {e}"
        except fastapi_mail.errors.ConnectionErrors as e:
            msg = f"连接失败: {e}"
        except Exception as e:
            msg = f"发送异常: {e}"

        res = EmailResult(
            ok=False,
            message=msg,
            error=str(e) if "e" in locals() else msg,
            subject=self._build_subject(subject),
            recipients=_normalize_recipients(recipients),
        )
        logger.error(f"邮件发送失败 | 接收人 - {res.recipients} | 主题 - {res.subject} | 原因 - {res.message}")
        return res

    # 便捷方法
    async def send_text(
        self,
        recipients: Union[str, EmailStr, Sequence[Union[str, EmailStr]]],
        subject: str,
        text: str,
        **kwargs: Any,
    ) -> EmailResult:
        return await self.send(
            subject=subject,
            recipients=recipients,
            subtype="plain",
            body=text,
            **kwargs,
        )

    async def send_html(
        self,
        recipients: Union[str, EmailStr, Sequence[Union[str, EmailStr]]],
        subject: str,
        html: str,
        **kwargs: Any,
    ) -> EmailResult:
        return await self.send(
            subject=subject,
            recipients=recipients,
            subtype="html",
            body=html,
            **kwargs,
        )

    async def send_template(
        self,
        recipients: Union[str, EmailStr, Sequence[Union[str, EmailStr]]],
        subject: str,
        template_name: str,
        context: Dict[str, Any],
        **kwargs: Any,
    ) -> EmailResult:
        return await self.send(
            subject=subject,
            recipients=recipients,
            template_name=template_name,
            template_context=context,
            **kwargs,
        )


# -------------------------
# 向下兼容的工具函数
# -------------------------
async def sys_send_mail(
    recipients: Union[str, List[str]],
    body: Union[str, Dict[str, Any]],
    subject: str,
    template_name: Optional[str] = None,
    config: Optional[Dict[str, Any]] = None,
    template_folder: Optional[DirectoryPath] = settings.BASE_TEMPLATES_DIR,
):
    """
    兼容你现有调用方式的函数。返回格式保持 {"code": 1|0, "message": "..."}。
    逻辑：
      - 若 body 为 dict 且提供 template_name => 作为模板上下文
      - 若 body 为 str => 作为 HTML body 发送
    """
    # 准备配置
    cfg_dict = config or settings.MAIL_CONFIG
    if not cfg_dict or not cfg_dict.get("MAIL_SERVER"):
        msg = "请配置邮件服务器"
        logger.error(f"邮件发送失败 | 原因 - {msg}")
        return {"code": 0, "message": msg}

    # 为模板发送设置模板目录
    if template_folder:
        cfg_dict["TEMPLATE_FOLDER"] = template_folder

    # 默认上下文（可自定义）
    default_ctx = {
        "sys_title": "测试标题",
        "sys_link": "https://www.baidu.com",
    }

    # 默认内联 logo（如果存在于模板目录）
    logo_path = Path(str(template_folder)) / "logo.png" if template_folder else None
    if logo_path and not logo_path.exists():
        logo_path = None

    service = EmailService(
        email_cfg=cfg_dict,
        template_folder=template_folder,
        default_context=default_ctx,
        subject_prefix="",  # 如需统一前缀可填，例如 "[系统通知] "
        inline_logo_path=logo_path,
    )

    if isinstance(body, dict):
        result = await service.send_template(
            recipients=recipients,
            subject=subject,
            template_name=template_name or "email.html",
            context=body,
        )
    else:
        result = await service.send_html(
            recipients=recipients,
            subject=subject,
            html=str(body),
        )

    return result.to_legacy()


# -------------------------
# Demo
# -------------------------
if __name__ == "__main__":

    async def _demo():
        # 模板发送
        res1 = await sys_send_mail(
            recipients="yuexiawyl@163.com",
            body={
                "title": "密码重置完成",
                "username": "test",
                "message": "新的密码为 asdasdsad",
            },
            subject="系统通知-重置密码",
            template_name="email.html",
        )
        print(res1)

        # 纯 HTML 发送
        # res2 = await sys_send_mail(
        #     recipients=["user1@example.com", "user2@example.com"],
        #     body="<h3>Hello</h3><p>这是一封测试邮件</p>",
        #     subject="系统通知-测试",
        # )
        # print(res2)

    asyncio.run(_demo())
